package com.atguigu.edu.realtime.common.base;

import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

public abstract class BaseApp {
    public abstract void handle(StreamExecutionEnvironment env , DataStreamSource<String> stream);
    /**
     * 抽取通用的代码， 作为一个模板， 进行复用。
     *
     * @param port
     *         测试环境下启动本地WebUI的端口，为了避免本地端口冲突，做出以下规定：
     *           （1）DIM层维度分流应用使用10001端口
     *           （2）DWD层应用程序按照在本文档中出现的先后顺序，端口从10011开始，自增1
     *           （3）DWS层应用程序按照在本文档中出现先后顺序，端口从10021开始，自增1
     * @param parallelism
     *             默认4个并行度， 对应Kafka主题的分区数， 也可以按照实际情况灵活调整。
     * @param ckAndGroupId
     *             消费Kafka主题时的消费者组ID和检查点路径的最后一级目录名称，二者取值相同。
     *             为Job主程序类名的下划线命名形式。如DimApp的该参数取值为dim_app。
     * @param topic
     *             消费的主题
     */
    public void start(Integer port ,Integer parallelism,String ckAndGroupId,String topic){
            // 设置HDFS用户
            System.setProperty(Constant.HDFS_USER_NAME_CONFIG, Constant.HDFS_USER_NAME_VALUE);
            // 配置
            Configuration conf = new Configuration();
            conf.setInteger(RestOptions.PORT, port);

            // 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
            // 调整并行度
            env.setParallelism(parallelism);

            // 2. 状态后端及检查点相关配置
            // 设置状态后端
            env.setStateBackend(new HashMapStateBackend());
            // 开启 checkpoint
            env.enableCheckpointing(5000 );
            // 设置 checkpoint 模式: 精准一次
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            // checkpoint 存储
            env.getCheckpointConfig().setCheckpointStorage(Constant.CK_PATH_PREFIX  + ckAndGroupId);
            // checkpoint 并发数
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
            // checkpoint 之间的最小间隔
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
            // checkpoint  的超时时间
            env.getCheckpointConfig().setCheckpointTimeout(20000);
            // job 取消时 checkpoint 保留策略
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);

            //3. 从数据源读取数据
            KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(topic, ckAndGroupId);

            DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");


            //4.对数据进行转换处理
            handle( env ,  stream );

            //5. 启动执行
            try {
                env.execute();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
    }
}
